package com.huatai.bi.mqtt;

import com.huatai.common.config.MinioAPI;
import com.huatai.bi.entity.MqMqttClientEntity;
import com.huatai.bi.utils.BeanUtil;
import com.huatai.bi.utils.SSLUtils;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.stereotype.Component;

import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSocketFactory;
import javax.net.ssl.TrustManager;
import javax.net.ssl.X509TrustManager;
import java.io.InputStream;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @program: itc-airport-cloud
 * @description: 创建mqtt链接
 * @author: songxiude
 * @create: 2024-07-15 15:18
 **/
@Slf4j
@Component
public class MqttClientConnect {


	private MqttClient mqttClient;

	/**
	 * 系统的mqtt客户端id
	 */
	private String mqttClientId;

	/**
	 * 客户端列表
	 */
	public static ConcurrentHashMap<String, MqttClientConnect> mqttClients = new ConcurrentHashMap();

	/**
	 * 客户端connect连接mqtt服务器
	 *
	 * @param userName     用户名
	 * @param passWord     密码
	 * @param mqttCallback 回调函数
	 **/
	public void setMqttClient(String host, String clientId, String userName, String passWord, boolean cleanSession, MqttCallback mqttCallback) throws MqttException {
		MqttConnectOptions options = mqttConnectOptions(host, clientId, userName, passWord, cleanSession);
		if (mqttCallback == null) {
			mqttClient.setCallback(new MqttClientCallback(mqttClientId));
		} else {
			mqttClient.setCallback(mqttCallback);
		}
		mqttClient.connect(options);
	}

	/**
	 * 客户端connect连接mqtt服务器
	 *
	 * @param mqttClientEntity
	 * @param mqttCallback
	 * @throws MqttException
	 */
	public void setMqttClient(MqMqttClientEntity mqttClientEntity, MqttCallback mqttCallback) throws MqttException {
		System.out.println("链接对象名称：" + mqttClientEntity.getMqttName());
		MqttConnectOptions options = mqttConnectOptions(mqttClientEntity);
		if (mqttCallback == null) {
			mqttClient.setCallback(new MqttClientCallback(mqttClientId));
		} else {
			mqttClient.setCallback(mqttCallback);
		}
		mqttClient.connect(options);
	}


	/**
	 * MQTT连接参数设置
	 */
	private MqttConnectOptions mqttConnectOptions(MqMqttClientEntity mqttClientEntity) throws MqttException {
		mqttClient = new MqttClient(mqttClientEntity.getMqttPrefix() + mqttClientEntity.getServerHost() + ":" + mqttClientEntity.getServerPort(), mqttClientEntity.getClientId(), new MemoryPersistence());
		MqttConnectOptions options = new MqttConnectOptions();
		options.setConnectionTimeout(30);///默认：30
		// 设置会话心跳时间
		options.setKeepAliveInterval(30);
		options.setAutomaticReconnect(true);//默认：false
//		options.setCleanSession(mqttClientEntity.getClientCleanSession());//默认：true
		options.setCleanSession(true);//默认：true
//		options.setCleanSession(false);//默认：true
		//options.setMaxReconnectDelay(1000);
		options.setHttpsHostnameVerificationEnabled(false); // 忽略主机名验证
		//判断链接类型
		if (mqttClientEntity.getClientType().intValue() == 1) {
			try {
				MinioAPI minioAPI = BeanUtil.getBean(MinioAPI.class);
				//SSLSocketFactory sslSocketFactory = SSLUtils.getSocketFactory("CA/caCert.pem", "CA/cert.txt", "CA/keys.txt", "");
				InputStream caFile = minioAPI.download(minioAPI.getBUCKET(), mqttClientEntity.getCaFile());
				InputStream ccFile = minioAPI.download(minioAPI.getBUCKET(), mqttClientEntity.getCcFile());
				InputStream ckFile = minioAPI.download(minioAPI.getBUCKET(), mqttClientEntity.getCkFile());
				SSLSocketFactory sslSocketFactory = SSLUtils.getSSLSocketFactory(caFile, ccFile, ckFile, "");

				options.setSocketFactory(sslSocketFactory);
			} catch (Exception e) {
				log.error("链接失败");
				e.printStackTrace();
			}
		} else if (mqttClientEntity.getClientType().intValue() == 2) {
			options.setUserName(mqttClientEntity.getUsername());
			options.setPassword(mqttClientEntity.getPassword().toCharArray());
		} else if (mqttClientEntity.getClientType().intValue() == 3) {
			options.setUserName(mqttClientEntity.getUsername());
			options.setPassword(mqttClientEntity.getPassword().toCharArray());

			try{
				// 创建信任所有证书的 SSLSocketFactory
				SSLContext sslContext = SSLContext.getInstance("TLS");
				sslContext.init(null, new TrustManager[]{new X509TrustManager() {
					@Override
					public void checkClientTrusted(java.security.cert.X509Certificate[] chain, String authType) {}
					@Override
					public void checkServerTrusted(java.security.cert.X509Certificate[] chain, String authType) {}
					@Override
					public java.security.cert.X509Certificate[] getAcceptedIssuers() {
						return new java.security.cert.X509Certificate[0];
					}
				}}, null);
				SSLSocketFactory sslSocketFactory = sslContext.getSocketFactory();
				options.setSocketFactory(sslSocketFactory);
			} catch (Exception e) {
				System.out.println("证书验证异常："+e.getMessage());
				e.printStackTrace();
			}
		}

		// 设置遗愿消息
		options.setWill("/lastWill", "device exception down".getBytes(), 1, false);
		return options;
	}

	/**
	 * MQTT连接参数设置
	 */
	private MqttConnectOptions mqttConnectOptions(String host, String clientId, String userName, String passWord, boolean cleanSession) throws MqttException {
		mqttClient = new MqttClient(host, clientId, new MemoryPersistence());
		MqttConnectOptions options = new MqttConnectOptions();
		options.setUserName(userName);
		options.setPassword(passWord.toCharArray());
		options.setConnectionTimeout(30);///默认：30
		options.setAutomaticReconnect(true);//默认：false
		options.setCleanSession(cleanSession);//默认：true
		//options.setKeepAliveInterval(20);//默认：60
		// 设置遗愿消息
		options.setWill("/lastWill", "device exception down".getBytes(), 1, false);
		return options;
	}

	/**
	 * 关闭MQTT连接
	 */
	public void close() throws MqttException {
//        mqttClient.close();
		mqttClient.disconnect();
	}

	/**
	 * 向某个主题发布消息 默认qos：1
	 *
	 * @param topic:发布的主题
	 * @param msg：发布的消息
	 */
	public void pub(String topic, String msg) throws MqttException {
		MqttMessage mqttMessage = new MqttMessage();
		//mqttMessage.setQos(2);
		mqttMessage.setPayload(msg.getBytes());
		MqttTopic mqttTopic = mqttClient.getTopic(topic);
		MqttDeliveryToken token = mqttTopic.publish(mqttMessage);
		token.waitForCompletion();
	}

	/**
	 * 向某个主题发布消息
	 *
	 * @param topic: 发布的主题
	 * @param msg:   发布的消息
	 * @param qos:   消息质量    Qos：0、1、2
	 */
	public void pub(String topic, String msg, int qos) throws MqttException {
		MqttMessage mqttMessage = new MqttMessage();
		mqttMessage.setQos(qos);
		mqttMessage.setPayload(msg.getBytes());
		MqttTopic mqttTopic = mqttClient.getTopic(topic);
		MqttDeliveryToken token = mqttTopic.publish(mqttMessage);
		token.waitForCompletion();
	}

	/**
	 * 订阅多个主题 ，此方法默认的的Qos等级为：1
	 *
	 * @param topic 主题
	 */
	public void sub(String[] topic) throws MqttException {
		mqttClient.subscribe(topic);
	}

	/**
	 * 订阅某一个主题，可携带Qos
	 *
	 * @param topic 所要订阅的主题
	 * @param qos   消息质量：0、1、2
	 */
	public void sub(String topic, int qos) throws MqttException {
		mqttClient.subscribe(topic, qos);
	}

	public String getMqttClientId() {
		return mqttClientId;
	}

	public void setMqttClientId(String mqttClientId) {
		this.mqttClientId = mqttClientId;
	}

}
